package com.example.service;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

@Component
public class Server implements ApplicationRunner {

    private static final Logger logger = LoggerFactory.getLogger(Server.class);

    @Autowired
    private ClientService clientService;

    private final ConcurrentHashMap<String, AtomicInteger> requestCounts = new ConcurrentHashMap<>();

    private final int f = 1;

    public void start() {
        Thread thread = new Thread(new connection());
        thread.start();
    }

    private class connection implements Runnable {
        @Override
        public void run() {
            try (Selector selector = Selector.open();
                    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {

                serverSocketChannel.socket().bind(new InetSocketAddress(7500));
                serverSocketChannel.configureBlocking(false);
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                logger.info("app启动 " + ":" + 7500);

                while (true) {
                    selector.select();
                    Set<SelectionKey> selectedKeys = selector.selectedKeys();
                    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                    while (keyIterator.hasNext()) {
                        SelectionKey key = keyIterator.next();

                        if (key.isAcceptable())
                            handleAccept(key, selector);
                        else if (key.isReadable())
                            handleRead(key);

                        keyIterator.remove();
                    }
                }
            } catch (IOException e) {
                System.out.println("监听异常" + e.getMessage());
            }
        }
    }

    private void handleAccept(SelectionKey key, Selector selector) throws IOException {
        ServerSocketChannel serverchannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverchannel.accept();
        clientChannel.configureBlocking(false);
        clientChannel.register(selector, SelectionKey.OP_READ);
        logger.info("客户端已接受连接来自: " + clientChannel.getRemoteAddress());
    }

    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
        int bytesRead = socketChannel.read(buffer);

        if (bytesRead > 0) {
            buffer.flip();
            byte[] data = new byte[buffer.remaining()];
            buffer.get(data);
            buffer.clear();
            String message = new String(data);
            Set<String> ids = parseRequestIds(message);
            ids.forEach(requestId -> {
                // 原子增加计数器（正确返回AtomicInteger）
                AtomicInteger count = requestCounts.compute(requestId,
                        (k, v) -> (v == null) ? new AtomicInteger(1) : new AtomicInteger(v.incrementAndGet()));

                // 获取当前计数值
                int currentCount = count.get();
                // 达到f+1个响应时触发
                if (currentCount >= f + 1) {
                    clientService.completeRequest(requestId);
                    requestCounts.remove(requestId); // 清理计数器
                }
            });
            socketChannel.close();
            if (!socketChannel.isConnected())
                logger.info("连接已关闭");
        }
    }

    // parseRequestIds方法
    private Set<String> parseRequestIds(String response) {
        Set<String> requestIds = new HashSet<>();
        try {
            JSONObject block = JSON.parseObject(response);
            JSONArray blockInfo = block.getJSONArray("blockInfo");
            if (blockInfo != null) {
                for (int i = 0; i < blockInfo.size(); i++) {
                    JSONObject entry = blockInfo.getJSONObject(i);
                    String requestId = entry.getString("requestId");
                    if (requestId != null)
                        requestIds.add(requestId);
                }
            }
        } catch (Exception e) {
            logger.warn("解析响应失败: " + e.getMessage());
        }
        return requestIds;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        clientService.sendConcurrentRequests(30, 2);
        start();
    }
}
